package realtime.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import realtime.util.DimUtil;
import realtime.util.DruidDSUtil;
import realtime.util.ThreadPoolUtil;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author MengX
 * @create 2023/3/25 16:36:36
 */
public abstract class  AsyncDIMFunction<T> extends RichAsyncFunction<T,T> implements AsyncFunDIM<T>{
    private final String tableName;
    private ThreadPoolExecutor threadPoolExecutor;
    public AsyncDIMFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
                String id = getKey(input);
//                String key = "DIM:" + tableName + id;
//                Jedis jedis = JedisUtil.getJedis();
//                String dimInfoStr = jedis.get(key);
//                JSONObject dimInfo = JSON.parseObject(dimInfoStr);
                JSONObject dimInfo = DimUtil.getDimInfo(phoenixConn, tableName, id);
//                if (dimInfo == null){
//                    jedis.setex(key,24*3600,dimInfo.toString());
//                    jedis.close();
//                }
                join(input,dimInfo);
                phoenixConn.close();
//                jedis.close();
                resultFuture.complete(Collections.singleton(input));
            }
        });

    }



    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("关联超时或者有误："+input);
    }
}
